package com.bingxu.flink.action;

import com.bingxu.flink.bean.MarketingUserBehavior;
import com.bingxu.flink.bean.UserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class Channel_Count {
    public static void main(String[] args) throws Exception {
        // 运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        AppMarketingSource source = new AppMarketingSource();
        DataStreamSource<MarketingUserBehavior> sourceDS = env.addSource(source);
        sourceDS
                .map(ele -> Tuple2.of(ele.getChannel() + ":" + ele.getBehavior(), 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(item -> item.f0)
                .sum(1)
                .print();

        env.execute();

    }

    public static class AppMarketingSource implements SourceFunction<MarketingUserBehavior> {

        private static boolean flag = true;

        private List<String> behaviorList = Arrays.asList("DOWNLOAD", "INSTALL", "UPDATE", "UNINSTALL");
        private List<String> channelList = Arrays.asList("XIAOMI", "HUAWEI", "OPPO", "APPSTORE", "VIVO");
        Random random = new Random();
        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (flag) {
                ctx.collect(new MarketingUserBehavior(
                        random.nextLong(),
                        behaviorList.get(random.nextInt(behaviorList.size())),
                        channelList.get(random.nextInt(channelList.size())),
                        System.currentTimeMillis()
                ));
                Thread.sleep(100);

            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}
